﻿

namespace Stee.CAP8.MessageBus.SampleProgram
{
    using System;
    using System.Text;
    using System.Threading;
    using Microsoft.Extensions.DependencyInjection;
    using NLog;
    using Stee.CAP8.MessageBus.Interface.Enums;
    using Stee.CAP8.MessageBus.Interface.Interfaces;
    using Stee.CAP8.MessageBus.Interface.Models;
    using Stee.CAP8.MessageBus.Kafka.Publisher;
    using Stee.CAP8.MessageBus.Kafka.ResourceLib;
    using Stee.CAP8.MessageBus.Kafka.Subscriber;
    using Stee.CAP8.MessageBus.SampleProgram.Models;

    class Program
    {
        private static readonly Logger Logger = LogManager.GetCurrentClassLogger();


        #region Topic Information

        private static readonly string topic = "first_topic";
        private static readonly string key = "first_key";
        private static readonly string groupID = "first_group";

        private static readonly string topic2 = "third_topic";
        private static readonly string key2 = "third_key";
        private static readonly string groupID2 = "third_group";

        private static readonly TopicName topicNameString = new TopicName(topic, key, MessageType.Event);
        private static readonly TopicName topicNameObject = new TopicName(topic2, key2, MessageType.Event);

        #endregion Topic Information


        #region Publisher & Subscriber

        private static ISubscriber<string> stringSubscriber;
        private static ISubscriber<Person> objectSubscriber;
        private static CancellationTokenSource stringSubscribeToken;
        private static CancellationTokenSource objectSubscribeToken;

        private static IPublisher<string> stringPublisher;
        private static IPublisher<Person> objectPublisher;
        private static CancellationTokenSource stringPublishToken;
        private static CancellationTokenSource objectPublishToken;

        #endregion Publisher & Subscriber


        #region Service Provider & Others

        private static ServiceProvider serviceProvider;

        #endregion Service Provider & Others


        static void Main(string[] args)
        {
            try
            {
                InitializeServiceProvider();

                SubscribeMessageTypeString();
                SubscribeMessageTypeObject();

                PublishMessageTypeString();
                PublishMessageTypeObject();

                //// Note: Loop in place to allow subscriber task to complete (threaded) before dispose.
                //// In production, subscriber will run indefinitely. For this demo, object message might be slow in consumption.
                for (int i = 0; i < 1000000000; i++)
                {
                    int j = 0;
                }

                DisposeSubscribers();
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
            }
        }

        /// <summary>
        /// Initializes the service provider in creating publisher & subscriber
        /// </summary>
        public static void InitializeServiceProvider()
        {
            IServiceCollection serviceCollection = new ServiceCollection();

            //// Add publisher scope
            serviceCollection.AddScoped<IPublisher<string>, PublisherImpl<string>>();
            serviceCollection.AddScoped<IPublisher<Person>, PublisherImpl<Person>>();

            //// Add subscriber scope
            serviceCollection.AddScoped<ISubscriber<string>, SubscriberImpl<string>>();
            serviceCollection.AddScoped<ISubscriber<Person>, SubscriberImpl<Person>>();

            //// Create service provider
            serviceProvider = serviceCollection.BuildServiceProvider();
        }


        #region Publish

        /// <summary>
        /// Publish messages of string data type
        /// </summary>
        private static void PublishMessageTypeString()
        {
            stringPublisher = serviceProvider.GetService<IPublisher<string>>();
            stringPublisher.Initialize(topicNameString, OnPublishMessage, OnPublisherError);

            if (stringPublisher.IsInitialized)
            {
                string message = RandomString(5);
                stringPublishToken = new CancellationTokenSource();
                bool isPublished = stringPublisher.Publish(message, stringPublishToken.Token);

                Console.WriteLine($"Publish random string => {message}");
            }

            stringPublishToken.Cancel();
            stringPublisher.Dispose();
        }

        /// <summary>
        /// Publish messages of object data type
        /// </summary>
        private static void PublishMessageTypeObject()
        {
            objectPublisher = serviceProvider.GetService<IPublisher<Person>>();
            objectPublisher.Initialize(topicNameObject, OnPublishPerson, OnPublisherError);

            if (objectPublisher.IsInitialized)
            {
                Person person = new Person(RandomString(5), RandomNumber(20, 50));
                objectPublishToken = new CancellationTokenSource();

                if (objectPublisher.Publish(person, objectPublishToken.Token))
                {
                    Console.WriteLine($"Publish Person object => {person}");
                }
                else
                {
                    Console.WriteLine($"[Error] Publish Person object failed => {person}");
                }
            }

            objectPublishToken.Cancel();
            objectPublisher.Dispose();
        }

        #endregion Publish


        #region Publish Handlers

        /// <summary>
        /// Mock example method name format, "On<Action><Type>".
        /// Publish response handler for string data type messages.
        /// </summary>
        /// <param name="response">Message publish outcome</param>
        /// <returns>Return TRUE if handled successfully else FALSE</returns>
        private static bool OnPublishMessage(PublishOutcome<string> response)
        {
            bool isHandled = false;

            if (response != null)
            {
                //// Handle / process outcome here
                isHandled = true;
            }

            return isHandled;
        }

        /// <summary>
        /// Mock example method name format, "On<Action><Type>".
        /// Publish response handler for Person data type messages.
        /// </summary>
        /// <param name="response">Message publish outcome</param>
        /// <returns>Return TRUE if handled successfully else FALSE</returns>
        private static bool OnPublishPerson(PublishOutcome<Person> response)
        {
            bool isHandled = false;

            if (response != null)
            {
                //// Handle / process outcome here
                isHandled = true;
            }

            return isHandled;
        }

        /// <summary>
        /// Mock example method name format, "On<Action><Response>".
        /// Publish error handler for broker publish error.
        /// </summary>
        /// <param name="errorCode">Publish error code</param>
        /// <param name="isFatal">Flag indicating fatal error</param>
        private static void OnPublisherError(string errorCode, bool isFatal)
        {
            if (string.IsNullOrEmpty(errorCode))
            {
                Logger.Warn($"Unexpected publisher error, no error code returned. Is fatal: {isFatal}");
            }
            else
            {
                Logger.Error($"Publisher error: {errorCode}, Is fatal: {isFatal}");
            }
        }

        #endregion Publish Handlers


        #region Subscribe

        /// <summary>
        /// Subscribe for messages of string data type
        /// </summary>
        private static void SubscribeMessageTypeString()
        {
            stringSubscriber = serviceProvider.GetService<ISubscriber<string>>();
            stringSubscriber.Initialize(topicNameString, OnMessageCreated, OnSubscriberError);

            if (stringSubscriber.IsInitialized)
            {
                stringSubscriber.Subscribe();
            }
        }

        /// <summary>
        /// Subscribe for messages of object data type
        /// </summary>
        private static void SubscribeMessageTypeObject()
        {
            objectSubscriber = serviceProvider.GetService<ISubscriber<Person>>();
            objectSubscriber.Initialize(topicNameObject, OnPersonCreated, OnSubscriberError);

            if (objectSubscriber.IsInitialized)
            {
                objectSubscriber.Subscribe();
            }
        }

        /// <summary>
        /// Dispose subscribers
        /// </summary>
        private static void DisposeSubscribers()
        {
            stringSubscriber.Dispose();
            objectSubscriber.Dispose();
        }

        #endregion Subscribe


        #region Subscribe Handlers

        /// <summary>
        /// Mock example method name format, "On<Type><Event/Command>".
        /// Handle message of string data type and commit.
        /// </summary>
        /// <param name="topicName">Topic name</param>
        /// <param name="message">Message, string type</param>
        /// <param name="brokerMessage">Raw broker message returned from subscription</param>
        /// <returns>Return TRUE if committed successfully else FALSE</returns>
        private static bool OnMessageCreated(TopicName topicName, string message, object brokerMessage)
        {
            try
            {
                if (topicName == null)
                {
                    throw new ArgumentNullException(nameof(topicName));
                }

                //// Handle / process messages here
                //// Consume message data type, string
                Console.WriteLine($"Consumed message <string>: {message}");
                Logger.Info($"{Kafka_Consumer.KAFKA_INFO_MESSAGE_CONSUMED}");

                //// Commit if consumed
                return stringSubscriber.Commit(brokerMessage);
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Mock example method name format, "On<Type><Event/Command>".
        /// Handle message of object data type and commit.
        /// </summary>
        /// <param name="topicName">Topic name</param>
        /// <param name="message">Message, object type</param>
        /// <param name="brokerMessage">Raw broker message returned from subscription</param>
        /// <returns>Return TRUE if committed successfully else FALSE</returns>
        private static bool OnPersonCreated(TopicName topicName, Person message, object brokerMessage)
        {
            try
            {
                if (topicName == null)
                {
                    throw new ArgumentNullException(nameof(topicName));
                }

                //// Consume data type, object
                Person person = message as Person;
                Console.WriteLine($"Consumed message <Person>: {person}");
                Logger.Info($"{Kafka_Consumer.KAFKA_INFO_MESSAGE_CONSUMED}");

                //// Commit if consumed
                return objectSubscriber.Commit(brokerMessage);
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Mock example method name format, "On<Action><Response>".
        /// Error handler for subscriber.
        /// </summary>
        /// <param name="errorCode">Broker error code</param>
        /// <param name="isFatal">Flag indicating fatal error</param>
        private static void OnSubscriberError(string errorCode, bool isFatal)
        {
            if (string.IsNullOrEmpty(errorCode))
            {
                Logger.Warn($"Unexpected subscriber error, no error code returned. Is fatal: {isFatal}");
            }
            else
            {
                Logger.Error($"Subscriber error: {errorCode}, Is fatal: {isFatal}");
            }
        }

        #endregion Subscribe Handlers


        #region Misc. Methods

        /// <summary>
        /// Generate random number
        /// </summary>
        /// <param name="min">Minimum value</param>
        /// <param name="max">Maximum value</param>
        /// <returns>Return random number</returns>
        public static int RandomNumber(int min = 0, int max = 100)
        {
            Random random = new Random();
            return random.Next(min, max);
        }

        /// <summary>
        /// Generate a random string with a given size
        /// </summary>
        /// <param name="size">String length</param>
        /// <returns>Return a new string</returns>
        public static string RandomString(int size = 5)
        {
            StringBuilder builder = new StringBuilder();
            Random random = new Random();
            char ch;

            for (int i = 0; i < size; i++)
            {
                ch = Convert.ToChar(Convert.ToInt32(Math.Floor(26 * random.NextDouble() + 65)));
                builder.Append(ch);
            }

            return builder.ToString();
        }

        #endregion Misc. Methods
    }
}
